package com.djhu.newscollector.crawel.imp.dispacth;

import com.djhu.newscollector.core.Holder.JobHolder;
import com.djhu.newscollector.core.dispatch.JobDispatcher;
import com.djhu.newscollector.core.job.BaseJob;
import com.djhu.newscollector.crawel.imp.holder.BaseJobHolder;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.spring.ActiveMQConnectionFactory;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.jms.core.JmsTemplate;

/**
 * Created by zw on 2018/2/11.
 */
public class JmsJobDispatcher implements JobDispatcher{

    private static Logger logger = LogManager.getLogger(BaseJobHolder.class.getName());

    private String url;

    private String queuename;

    private JmsTemplate jmsTemplate;

    public JmsJobDispatcher(String url,String queuename){
        this.url = url;
        this.queuename = queuename;
        init();
    }

    public void init(){
        ActiveMQConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory();
        mqConnectionFactory.setTrustAllPackages(true);
        mqConnectionFactory.setBrokerURL(url);
        mqConnectionFactory.setMaxThreadPoolSize(10);

        ActiveMQQueue queue = new ActiveMQQueue(queuename);

        jmsTemplate = new JmsTemplate(mqConnectionFactory);
        jmsTemplate.setDefaultDestination(queue);
    }

    @Override
    public BaseJob dispatcher() {
        Object msg = jmsTemplate.receiveAndConvert();
        if (msg instanceof BaseJob){
            return (BaseJob)msg;
        }else {
            logger.info("receive a msg is not job!!");
            return null;
        }
    }

    @Override
    public Boolean retJob(BaseJob job) {
        jmsTemplate.convertAndSend(job);
        return true;
    }

}